package com.techsun.handler;

import com.alibaba.fastjson.JSONObject;
import com.techsun.utils.MysqlUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.http.HTTPSourceHandler;
import org.apache.log4j.Logger;

import javax.servlet.http.HttpServletRequest;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.*;

/**
 * 将不同租户数据发往不同 top , 规则: retailevent 和 customer 加上 租户所对应的环境id的后缀
 */
public class TenantByEnvHTTPHandler implements HTTPSourceHandler {
    Logger logger = Logger.getLogger(TenantByEnvHTTPHandler.class);
    public static final String charset = "UTF-8";
    private final String[] devEnvironment = new String[]{"dev", ""};
    private final String[] uatEnvironment = new String[]{"uat", ""};
    private final String[] prodEnvironment = new String[]{"prod", ""};
    static Map<String, String> topicMap = new HashMap<>();

    @Override
    public void configure(Context context) {
        devEnvironment[1] = context.getString("devConnectionURL", "");
        uatEnvironment[1] = context.getString("uatConnectionURL", "");
        prodEnvironment[1] = context.getString("prodConnectionURL", "");
        Long pollIntervalSecond = context.getLong("pollIntervalSecond", 60L);
        initTopicMap();
        refreshTopicMap(pollIntervalSecond);
    }

    private void refreshTopicMap(Long pollIntervalSecond) {
        new Thread(() -> {
            while (true) {
                this.initTopicMap();
                try {
                    Thread.sleep(pollIntervalSecond * 1000);
                } catch (Exception e) {
                    logger.error("refresh topic map has error: ", e);
                }
            }
        }).start();
    }

    private void initTopicMap() {
        for (String[] environment : Arrays.asList(devEnvironment, uatEnvironment, prodEnvironment)) {
            if (!"".equals(environment[1])) {
                MysqlUtil mysqlUtil = null;
                try {
                    mysqlUtil = new MysqlUtil(environment[1]);
                    List<JSONObject> select = mysqlUtil.select("SELECT tenant_id FROM tenant_info;");
                    for (JSONObject jsonObject : select) {
                        topicMap.put(jsonObject.getString("tenant_id"), environment[0]);
                    }
                    mysqlUtil.close();
                } catch (Exception e) {
                    logger.error("init topic map has error: ", e);
                } finally {
                    if (mysqlUtil != null)
                        mysqlUtil.close();
                }
            }
        }
    }

    @Override
    public List<Event> getEvents(HttpServletRequest request) {
        StringBuilder sb = new StringBuilder();
        List<Event> events = new LinkedList<>();
        try {
            if (request.getMethod().equalsIgnoreCase("POST")) {
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(request.getInputStream(), charset));
                String inputStr;
                while ((inputStr = bufferedReader.readLine()) != null) {
                    sb.append(inputStr);
                }
                logger.info(sb.toString());

                JSONObject jsonObject = JSONObject.parseObject(sb.toString());
                // 获取租户id
                String environmentStr = topicMap.getOrDefault(jsonObject.getString("project"), "error");
                if ("error".equals(environmentStr)) {
                    logger.error(String.format("this request: [%s] does not match any environment: [%s]", sb, topicMap));
                }
                String customerTopic = "customer_" + environmentStr;
                String retailEventTopic = "retailevent_" + environmentStr;
                if (!jsonObject.containsKey("posted_time")) {
                    jsonObject.put("posted_time", System.currentTimeMillis());
                }
                if (jsonObject.getString("event_type").equalsIgnoreCase("track_signup")) {
                    events.add(EventBuilder.withBody(jsonObject.toString().getBytes(charset), Collections.singletonMap("topic", customerTopic)));
                }
                events.add(EventBuilder.withBody(jsonObject.toString().getBytes(charset), Collections.singletonMap("topic", retailEventTopic)));
                return events;
            }
        } catch (Exception e) {
            logger.error(String.format("this request: [%s] parse data exception: [%s]", sb, e));
        }
        return events;
    }


}
